using System.Collections.Concurrent;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using PHMEE.VibrationDataGenerator.Data;

namespace PHMEE.VibrationDataGenerator.Services;

public class MqttClientPool
{
    private readonly MqttSettings _mqttSettings;
    
    private readonly ILogger<MqttClientPool> _logger;
    private readonly MqttFactory _mqttFactory;
    
    private readonly ConcurrentBag<IManagedMqttClient> _clients;
    private readonly int _maxSize;
    private int _currentSize;
    private readonly SemaphoreSlim _semaphore;

    public MqttClientPool(int maxSize, IOptions<MqttSettings> mqttSettings, ILogger<MqttClientPool> logger)
    {
        _mqttSettings = mqttSettings.Value;
        
        _mqttFactory = new MqttFactory();
        
        _clients = new ConcurrentBag<IManagedMqttClient>();
        _maxSize = maxSize;
        _logger = logger;
        _currentSize = 0;
        _semaphore = new SemaphoreSlim(maxSize, maxSize);
    }
    
    private ManagedMqttClientOptions CreateManagedMqttClientOptions()
    {
        return new ManagedMqttClientOptionsBuilder()
            .WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
            .WithClientOptions(new MqttClientOptionsBuilder()
                .WithTcpServer(_mqttSettings.Host, _mqttSettings.Port)
                .WithCredentials(_mqttSettings.UserName, _mqttSettings.Password)
                .WithCleanSession(true)
                .WithClientId(Guid.NewGuid().ToString())
                .Build())
            .Build();
    }

    public async Task<MqttClientWrapper> GetClientAsync()
    {
        await _semaphore.WaitAsync();
        try
        {
            while (true)
            {
                if (_clients.TryTake(out var client))
                {
                    if (IsClientHealthy(client))
                    {
                        return new MqttClientWrapper(this, client);
                    }
                    else
                    {
                        // 后台修复客户端
                        RepairClientInBackground(client);
                    }
                }

                if (_currentSize < _maxSize)
                {
                    Interlocked.Increment(ref _currentSize);
                    var newClient = _mqttFactory.CreateManagedMqttClient();
                    var managedOptions = CreateManagedMqttClientOptions();
                    await newClient.StartAsync(managedOptions);
                    
                    // 在代码中，新创建的客户端在返回给调用方后并没有立即添加到连接池中。
                    // 这是有意为之的设计，目的是为了确保每个客户端在被创建后首先能够被使用，然后在调用方使用完毕后再通过 ReleaseClient 方法将其归还到池中
                    return new MqttClientWrapper(this, newClient);
                }

                await Task.Delay(100); // 等待片刻再重试获取客户端
            }
        }
        catch (Exception ex)
        {
            _semaphore.Release();
            _logger.LogError("Failed to get client: {Message}", ex.Message);
            throw;
        }
    }

    private bool IsClientHealthy(IManagedMqttClient client)
    {
        if (!client.IsConnected || !client.IsStarted)
        {
            return false;
        }

        return true; 
    }

    private void RepairClientInBackground(IManagedMqttClient client)
    {
        Task.Run(() =>
        {
            try
            {
                client.StopAsync();
                client.StartAsync(client.Options);
                _clients.Add(client);
            }
            catch (Exception ex)
            {
                _logger.LogError("Failed to repair client: {Message}", ex.Message);
                Interlocked.Decrement(ref _currentSize);
            }
        });
    }

    public void ReleaseClient(IManagedMqttClient client)
    {
        try
        {
            _clients.Add(client);
        }
        finally
        {
            _semaphore.Release();
        }
    }
}